package com.boot.admin.modules.sys.ai.service;

import cn.hutool.core.util.IdUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.boot.admin.modules.sys.ai.config.XfyunAiConfig;
import com.boot.admin.modules.sys.ai.model.entity.AiConversation;
import com.boot.admin.modules.sys.ai.model.entity.AiMessage;
import com.boot.admin.modules.sys.ai.model.entity.AiRole;
import com.boot.admin.modules.sys.ai.model.vo.AiMessageVO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import okhttp3.*;
import org.springframework.stereotype.Service;
import org.springframework.util.Base64Utils;
import org.springframework.web.socket.BinaryMessage;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.io.ByteArrayOutputStream;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;

import static com.baomidou.mybatisplus.extension.toolkit.Db.getById;
import static com.baomidou.mybatisplus.extension.toolkit.Db.updateById;

/**
 * 实时语音对话服务 - 完整优化版
 *
 * @author 李飞洋
 * @date 2025/9/24
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class RealtimeVoiceService {

    private final AiAsrService aiAsrService;
    private final AiTtsService aiTtsService;
    private final LocalSseLlmService localSseLlmService;
    private final AiRoleService aiRoleService;
    private final AiMessageService aiMessageService;
    private final AiConversationService aiConversationService;
    private final StreamingWebSocketService streamingWebSocketService;
    private final XfyunAiConfig xfyunAiConfig;
    private final TextBufferService textBufferService;

    // 存储每个会话的语音流缓冲区
    private final ConcurrentHashMap<String, VoiceStreamBuffer> sessionBuffers = new ConcurrentHashMap<>();

    // 存储每个对话的状态
    private final ConcurrentHashMap<String, ConversationState> conversationStates = new ConcurrentHashMap<>();

    // 存储每个对话的ASR连接
    private final ConcurrentHashMap<String, AsrConnection> activeAsrConnections = new ConcurrentHashMap<>();

    // WebSocket会话发送锁
    private final ConcurrentHashMap<String, Object> sessionSendLocks = new ConcurrentHashMap<>();

    // 定时任务执行器
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5);

    // OkHttpClient
    private final OkHttpClient httpClient = new OkHttpClient.Builder()
            .connectTimeout(15, TimeUnit.SECONDS)
            .writeTimeout(60, TimeUnit.SECONDS)
            .readTimeout(60, TimeUnit.SECONDS)
            .build();

    /**
     * 获取WebSocket会话发送锁
     */
    private Object getSessionLock(WebSocketSession session) {
        return sessionSendLocks.computeIfAbsent(session.getId(), k -> new Object());
    }

    /**
     * 处理实时语音流数据
     */
    public void handleVoiceStream(WebSocketSession session, BinaryMessage message, String conversationId) {
        try {
            byte[] audioData = message.getPayload().array();

            VoiceStreamBuffer buffer = sessionBuffers.computeIfAbsent(
                    session.getId(),
                    k -> new VoiceStreamBuffer(session, conversationId)
            );

            buffer.addAudioData(audioData);

            ConversationState state = conversationStates.get(conversationId);
            if (state != null && state.isRecording() && buffer.shouldProcess()) {
                processAudioSegmentStreaming(buffer, state, session);
            }

        } catch (Exception e) {
            log.error("处理实时语音流失败", e);
            sendErrorToClient(session, "音频流处理失败: " + e.getMessage());
        }
    }

    /**
     * 流式音频处理 - 改进版，防止重复发送
     */
    private void processAudioSegmentStreaming(VoiceStreamBuffer buffer, ConversationState state, WebSocketSession session) {
        try {
            byte[] audioSegment = buffer.getAndClearBuffer();

            if (audioSegment.length < 2048) {
                log.debug("音频片段过短: {} bytes，跳过", audioSegment.length);
                return;
            }

            boolean isActive = isVoiceActive(audioSegment);
            if (!isActive) {
                log.debug("检测到静音片段，跳过处理");
                return;
            }

            String conversationId = buffer.getConversationId();
            AsrConnection asrConn = activeAsrConnections.get(conversationId);

            if (asrConn == null || !asrConn.isActive() || asrConn.getWebSocket() == null) {
                log.warn("ASR连接不可用 - 对话: {}", conversationId);
                return;
            }

            log.debug("发送音频帧到ASR - 大小: {} bytes, 对话: {}", audioSegment.length, conversationId);

            // 只发送音频数据帧（status=1，中间帧）
            sendAudioFrame(asrConn.getWebSocket(), audioSegment);

        } catch (Exception e) {
            log.error("处理音频片段失败", e);
        }
    }

    /**
     * 开始录音 - 建立持久ASR连接
     */
    public void startRecording(String conversationId, WebSocketSession session) {
        ConversationState state = conversationStates.get(conversationId);
        if (state != null) {
            state.setRecording(true);
            state.clearAsrText();

            // 建立ASR连接
            AsrConnection asrConn = new AsrConnection();
            activeAsrConnections.put(conversationId, asrConn);

            // 初始化讯飞ASR WebSocket连接
            initAsrConnection(conversationId, session, asrConn);

            log.info("开始录音并建立ASR连接 - 对话: {}", conversationId);
        }
    }

    /**
     * 初始化ASR连接
     */
    private void initAsrConnection(String conversationId, WebSocketSession session, AsrConnection asrConn) {
        CompletableFuture.runAsync(() -> {
            try {
                String authUrl = buildAsrAuthUrl();
                Request request = new Request.Builder().url(authUrl).build();

                WebSocket ws = httpClient.newWebSocket(request, new WebSocketListener() {
                    @Override
                    public void onOpen(WebSocket webSocket, Response response) {
                        log.debug("ASR连接已建立 - 对话: {}", conversationId);
                        asrConn.setWebSocket(webSocket);
                        try {
                            sendAsrConfig(webSocket);
                        } catch (Exception e) {
                            log.error("发送ASR配置失败", e);
                        }
                    }

                    @Override
                    public void onMessage(WebSocket webSocket, String text) {
                        try {
                            JSONObject response = JSONUtil.parseObj(text);
                            int code = response.getInt("code", 0);

                            if (code != 0) {
                                String message = response.getStr("message", "未知错误");
                                log.error("ASR返回错误: code={}, message={}", code, message);
                                sendErrorToClient(session, "ASR识别错误: " + message);
                                return;
                            }

                            JSONObject data = response.getJSONObject("data");
                            if (data != null) {
                                String result = data.getStr("result");
                                int status = data.getInt("status", 0);

                                // 处理中间结果
                                if (result != null && !result.trim().isEmpty()) {
                                    String recognizedText = parseAsrResult(result);
                                    if (recognizedText != null && !recognizedText.trim().isEmpty()) {
                                        asrConn.setText(recognizedText);

                                        // 只在非最终结果时发送给前端显示
                                        if (status != 2) {
                                            sendAsrResultToClient(session, conversationId, recognizedText, false);
                                            log.debug("ASR中间结果: '{}'", recognizedText);
                                        } else {
                                            // status=2时只记录日志，不发送，由handleRecordingStop统一处理
                                            log.info("ASR识别完成，最终结果: '{}'", recognizedText);
                                        }
                                    }
                                }
                            }
                        } catch (Exception e) {
                            log.error("解析ASR消息失败", e);
                        }
                    }

                    @Override
                    public void onFailure(WebSocket webSocket, Throwable t, Response response) {
                        log.error("ASR连接失败", t);
                        sendErrorToClient(session, "ASR连接失败: " + t.getMessage());
                    }

                    @Override
                    public void onClosed(WebSocket webSocket, int code, String reason) {
                        log.debug("ASR连接已关闭: {} - {}", code, reason);
                    }
                });

            } catch (Exception e) {
                log.error("初始化ASR连接失败", e);
                sendErrorToClient(session, "初始化ASR连接失败: " + e.getMessage());
            }
        });
    }

    /**
     * 处理录音停止 - 发送结束帧并触发LLM
     */
    public void handleRecordingStop(WebSocketSession session, String conversationId, String roleId) {
        try {
            ConversationState state = conversationStates.get(conversationId);
            AsrConnection asrConn = activeAsrConnections.remove(conversationId);

            if (asrConn != null && asrConn.getWebSocket() != null) {
                sendEndFrame(asrConn.getWebSocket());
                Thread.sleep(2000);

                String finalText = asrConn.getFinalText();
                asrConn.close();

                if (finalText.isEmpty()) {
                    log.warn("录音结束但未识别到内容");
                    sendAsrResultToClient(session, conversationId, "", true);
                    if (state != null) {
                        state.setRecording(false);
                        state.clearAsrText();
                    }
                    return;
                }

                log.info("录音结束，最终识别结果: '{}'", finalText);
                sendAsrResultToClient(session, conversationId, finalText, true);

                processUserVoiceInputWithTtsBuffer(conversationId, finalText, session, roleId);
            }

            if (state != null) {
                state.setRecording(false);
                state.clearAsrText();
            }

        } catch (Exception e) {
            log.error("处理录音停止失败", e);
            sendErrorToClient(session, "处理录音停止失败: " + e.getMessage());
        }
    }

    /**
     * 处理用户语音输入 - 使用智能文本缓冲优化TTS
     */
    private void processUserVoiceInputWithTtsBuffer(String conversationId, String userText,
                                                    WebSocketSession session, String roleId) {
        try {
            AiRole role = aiRoleService.getRoleById(roleId);
            if (role == null) {
                sendErrorToClient(session, "角色不存在");
                return;
            }

            List<AiMessageVO> history = aiMessageService.getRecentMessages(conversationId, 10);
            Integer nextIndex = aiMessageService.getNextMessageIndex(conversationId);

            // 保存用户消息
            String userMessageId = IdUtil.simpleUUID();
            AiMessage userMessage = new AiMessage();
            userMessage.setId(userMessageId);
            userMessage.setConversationId(conversationId);
            userMessage.setSenderType(1);
            userMessage.setMessageType(1);
            userMessage.setContent(userText);
            userMessage.setMessageIndex(nextIndex);
            userMessage.setCreated(LocalDateTime.now());

            aiMessageService.save(userMessage);
            log.info("保存用户消息 - ID: {}, Index: {}, Content: {}", userMessageId, nextIndex, userText);

            sendUserInputCompleteToClient(session, conversationId, userText);

            String aiMessageId = IdUtil.simpleUUID();

            // 初始化智能文本缓冲区
            TextBufferService.ConversationBuffer textBuffer =
                    textBufferService.initializeBuffer(conversationId, aiMessageId);

            // TTS调用去重控制
            final Set<String> processedSentences = ConcurrentHashMap.newKeySet();

            CompletableFuture.runAsync(() -> {
                try {
                    localSseLlmService.generateResponseStreaming(
                            role, history, userText, false,
                            new AiConversationService.StreamingCallback() {
                                @Override
                                public void onTextChunk(String chunk) {
                                    try {
                                        // 发送文本chunk给客户端
                                        sendTextChunkToClient(session, conversationId, aiMessageId, chunk);

                                        // 使用智能文本缓冲服务处理文本
                                        TextBufferService.TextProcessResult result =
                                                textBufferService.processTextChunk(conversationId, aiMessageId, chunk);

                                        // 处理完成的句子
                                        if (result.hasNewSentence()) {
                                            for (String completedSentence : result.getCompletedSentences()) {
                                                // 去重检查：避免处理相同的句子
                                                String sentenceHash = Integer.toString(completedSentence.hashCode());
                                                if (processedSentences.add(sentenceHash)) {
                                                    // 异步调用TTS处理每个完成的句子
                                                    CompletableFuture.runAsync(() -> {
                                                        generateAndSendAudioChunkWithRetry(session, conversationId,
                                                                aiMessageId, completedSentence, role, 3);
                                                    });

                                                    log.debug("处理完成句子: '{}' (Hash: {})",
                                                            completedSentence.substring(0, Math.min(completedSentence.length(), 20)) + "...",
                                                            sentenceHash);
                                                } else {
                                                    log.debug("跳过重复句子: '{}'",
                                                            completedSentence.substring(0, Math.min(completedSentence.length(), 20)) + "...");
                                                }
                                            }
                                        }

                                    } catch (Exception e) {
                                        log.error("处理AI文本chunk失败", e);
                                    }
                                }

                                @Override
                                public void onComplete() {
                                    try {
                                        // 标记流结束并处理剩余文本
                                        TextBufferService.TextProcessResult finalResult =
                                                textBufferService.markStreamCompleted(conversationId, aiMessageId);

                                        // 处理最后剩余的句子
                                        if (finalResult.hasNewSentence()) {
                                            for (String finalSentence : finalResult.getCompletedSentences()) {
                                                String sentenceHash = Integer.toString(finalSentence.hashCode());
                                                if (processedSentences.add(sentenceHash)) {
                                                    CompletableFuture.runAsync(() -> {
                                                        generateAndSendAudioChunkWithRetry(session, conversationId,
                                                                aiMessageId, finalSentence, role, 3);
                                                    });

                                                    log.debug("处理最终句子: '{}'",
                                                            finalSentence.substring(0, Math.min(finalSentence.length(), 20)) + "...");
                                                }
                                            }
                                        }

                                        // 获取完整内容并持久化到数据库
                                        String completeContent = textBufferService.getCompleteContent(conversationId, aiMessageId);

                                        if (completeContent != null && !completeContent.trim().isEmpty()) {
                                            // 保存AI消息到数据库
                                            AiMessage aiMessage = new AiMessage();
                                            aiMessage.setId(aiMessageId);
                                            aiMessage.setConversationId(conversationId);
                                            aiMessage.setSenderType(2); // AI
                                            aiMessage.setMessageType(1); // 文本
                                            aiMessage.setContent(completeContent);
                                            aiMessage.setMessageIndex(nextIndex + 1);
                                            aiMessage.setCreated(LocalDateTime.now());

                                            aiMessageService.save(aiMessage);

                                            log.info("保存AI消息到数据库 - ID: {}, Index: {}, Content长度: {}",
                                                    aiMessageId, nextIndex + 1, completeContent.length());

                                            // 更新对话最后消息时间
                                            AiConversation conversation = getById(conversationId, AiConversation.class);
                                            if (conversation != null) {
                                                conversation.setLastMessageTime(LocalDateTime.now());
                                                updateById(conversation);
                                            }
                                        } else {
                                            log.warn("AI消息内容为空，跳过数据库保存 - MessageId: {}", aiMessageId);
                                        }

                                        // 发送AI响应完成信号
                                        sendAiResponseCompleteToClient(session, conversationId, aiMessageId);

                                        // 发送音频完成信号
                                        streamingWebSocketService.sendAudioComplete(conversationId, aiMessageId);

                                        // 清理文本缓冲区
                                        textBufferService.cleanupBuffer(conversationId, aiMessageId);

                                        log.info("AI语音对话处理完成 - 对话: {}, 消息: {}, 处理句子数: {}",
                                                conversationId, aiMessageId, processedSentences.size());

                                    } catch (Exception e) {
                                        log.error("完成AI回复处理失败", e);
                                        sendErrorToClient(session, "完成AI回复处理失败: " + e.getMessage());
                                    }
                                }

                                @Override
                                public void onError(String error) {
                                    log.error("LLM生成失败: {}", error);
                                    sendErrorToClient(session, "AI回复生成失败: " + error);

                                    // 清理文本缓冲区
                                    textBufferService.cleanupBuffer(conversationId, aiMessageId);
                                }
                            }
                    );
                } catch (Exception e) {
                    log.error("LLM流式处理异常", e);
                    sendErrorToClient(session, "LLM处理异常: " + e.getMessage());

                    // 清理文本缓冲区
                    textBufferService.cleanupBuffer(conversationId, aiMessageId);
                }
            });

        } catch (Exception e) {
            log.error("处理用户语音输入失败", e);
            sendErrorToClient(session, "处理用户输入失败: " + e.getMessage());
        }
    }

    /**
     * 生成并发送音频chunk（带重试机制）
     */
    private void generateAndSendAudioChunkWithRetry(WebSocketSession session, String conversationId,
                                                    String messageId, String textChunk, AiRole role, int maxRetries) {
        for (int attempt = 1; attempt <= maxRetries; attempt++) {
            try {
                if (textChunk == null || textChunk.trim().length() < 2) {
                    log.debug("文本过短，跳过TTS处理: {}", textChunk);
                    return;
                }

                // 预处理文本
                String cleanText = preprocessTextForTts(textChunk);
                if (cleanText.trim().length() < 2) {
                    log.debug("文本预处理后过短，跳过TTS: {}", textChunk);
                    return;
                }

                log.debug("开始TTS处理 (尝试 {}/{}) - 文本: '{}'",
                        attempt, maxRetries, cleanText.substring(0, Math.min(cleanText.length(), 30)) + "...");

                // 调用TTS服务
                com.boot.admin.base.R<byte[]> ttsResult = aiTtsService.generateAudioChunk(cleanText, role);

                if (ttsResult.isSuccess()) {
                    byte[] audioChunk = ttsResult.getResult();
                    if (audioChunk != null && audioChunk.length > 0) {
                        // 发送音频chunk给客户端
                        sendAudioChunkToClient(session, conversationId, messageId, audioChunk);

                        log.debug("TTS成功 - 音频大小: {} bytes, 文本: '{}'",
                                audioChunk.length, cleanText.substring(0, Math.min(cleanText.length(), 20)) + "...");
                        return; // 成功，退出重试循环
                    } else {
                        log.warn("TTS返回空音频数据 (尝试 {}/{})", attempt, maxRetries);
                    }
                } else {
                    log.warn("TTS生成失败 (尝试 {}/{}): {}", attempt, maxRetries, ttsResult.getMessage());
                }

                // 如果不是最后一次尝试，等待一段时间再重试
                if (attempt < maxRetries) {
                    Thread.sleep(500 * attempt); // 递增等待时间
                }

            } catch (Exception e) {
                log.error("TTS处理异常 (尝试 {}/{})", attempt, maxRetries, e);

                if (attempt < maxRetries) {
                    try {
                        Thread.sleep(1000 * attempt); // 异常时等待更长时间
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
        }

        // 所有重试都失败
        log.error("TTS处理最终失败，已重试 {} 次 - 文本: '{}'",
                maxRetries, textChunk.substring(0, Math.min(textChunk.length(), 30)) + "...");

        // 可选：发送错误通知给客户端（但不中断整个流程）
        try {
            streamingWebSocketService.sendError(conversationId,
                    "音频生成失败，但文本正常显示：" + textChunk.substring(0, Math.min(textChunk.length(), 20)) + "...");
        } catch (Exception e) {
            log.error("发送TTS错误通知失败", e);
        }
    }

    /**
     * 生成并发送音频chunk（优化版：文本预处理和分段）
     */
    private void generateAndSendAudioChunk(WebSocketSession session, String conversationId,
                                           String messageId, String textChunk, AiRole role) {
        try {
            if (textChunk.trim().length() < 2) {
                return;
            }

            String cleanText = preprocessTextForTts(textChunk);

            if (cleanText.trim().length() < 2) {
                log.debug("文本预处理后过短，跳过TTS: {}", textChunk);
                return;
            }

            if (cleanText.length() > 50) {
                List<String> segments = splitTextIntoSegments(cleanText, 30);
                for (String segment : segments) {
                    generateSingleAudioChunk(session, conversationId, messageId, segment, role);
                }
            } else {
                generateSingleAudioChunk(session, conversationId, messageId, cleanText, role);
            }

        } catch (Exception e) {
            log.error("生成音频chunk失败", e);
        }
    }

    /**
     * TTS文本预处理
     */
    private String preprocessTextForTts(String text) {
        if (text == null) {
            return "";
        }

        String cleaned = text
                .replaceAll("[¡¿]", "")
                .replaceAll("[（）()]", "")
                .replaceAll("\"", "")
                .replaceAll("\\s+", " ")
                .trim();

        return cleaned;
    }

    /**
     * 将长文本分段
     */
    private List<String> splitTextIntoSegments(String text, int maxLength) {
        List<String> segments = new ArrayList<>();

        String[] sentences = text.split("[。！？，,.!?;；]");

        StringBuilder currentSegment = new StringBuilder();
        for (String sentence : sentences) {
            if (sentence.trim().isEmpty()) {
                continue;
            }

            if (currentSegment.length() + sentence.length() > maxLength && currentSegment.length() > 0) {
                segments.add(currentSegment.toString().trim());
                currentSegment.setLength(0);
            }

            currentSegment.append(sentence).append("。");
        }

        if (currentSegment.length() > 0) {
            segments.add(currentSegment.toString().trim());
        }

        if (segments.isEmpty()) {
            segments.add(text);
        }

        return segments;
    }

    /**
     * 生成单个音频chunk（带重试）
     */
    private void generateSingleAudioChunk(WebSocketSession session, String conversationId,
                                          String messageId, String textChunk, AiRole role) {
        int maxRetries = 2;
        for (int i = 0; i < maxRetries; i++) {
            try {
                com.boot.admin.base.R<byte[]> ttsResult = aiTtsService.generateAudioChunk(textChunk, role);

                if (ttsResult.isSuccess()) {
                    byte[] audioChunk = ttsResult.getResult();
                    if (audioChunk != null && audioChunk.length > 0) {
                        sendAudioChunkToClient(session, conversationId, messageId, audioChunk);
                        log.debug("发送音频chunk: {} bytes, 文本: {}", audioChunk.length, textChunk);
                    }
                    return;
                } else {
                    log.warn("TTS生成失败（尝试 {}/{}）: {}, 文本: {}",
                            i + 1, maxRetries, ttsResult.getMessage(), textChunk);
                    if (i < maxRetries - 1) {
                        Thread.sleep(500);
                    }
                }
            } catch (Exception e) {
                log.error("TTS处理异常（尝试 {}/{}）", i + 1, maxRetries, e);
                if (i == maxRetries - 1) {
                    break;
                }
            }
        }

        log.error("TTS生成最终失败，文本: {}", textChunk);
    }

    /**
     * 语音活动检测
     */
    private boolean isVoiceActive(byte[] audioData) {
        if (audioData.length == 0) {
            return false;
        }

        if (audioData.length % 2 != 0) {
            log.warn("音频数据长度不是偶数，可能不是16位PCM格式");
            return false;
        }

        double energy = 0.0;
        int zeroCrossings = 0;
        short prevSample = 0;

        for (int i = 0; i < audioData.length - 1; i += 2) {
            short sample = (short) ((audioData[i] & 0xFF) | ((audioData[i + 1] & 0xFF) << 8));
            energy += sample * sample;

            if (i > 0 && ((prevSample >= 0 && sample < 0) || (prevSample < 0 && sample >= 0))) {
                zeroCrossings++;
            }
            prevSample = sample;
        }

        energy /= (audioData.length / 2);
        double zcr = (double) zeroCrossings / (audioData.length / 2) * 1000;

        double energyThreshold = 3000.0;
        double zcrThreshold = 20.0;

        boolean isActive = energy > energyThreshold && zcr > zcrThreshold;
        log.debug("VAD检测 - 能量: {}, 零交叉率: {}, 活跃: {}", energy, zcr, isActive);

        return isActive;
    }

    /**
     * 初始化对话状态
     */
    public void initConversationState(String conversationId, String roleId, String userId) {
        ConversationState state = new ConversationState();
        state.setConversationId(conversationId);
        state.setRoleId(roleId);
        state.setUserId(userId);
        state.setActive(true);

        conversationStates.put(conversationId, state);
        log.info("初始化对话状态: {}", conversationId);

        scheduler.schedule(() -> cleanupExpiredConversation(conversationId), 30, TimeUnit.MINUTES);
    }

    /**
     * 清理会话资源
     */
    public void cleanupSession(String sessionId) {
        VoiceStreamBuffer buffer = sessionBuffers.remove(sessionId);
        sessionSendLocks.remove(sessionId);
        if (buffer != null) {
            buffer.cleanup();
            log.info("清理会话资源: {}", sessionId);
        }
    }

    /**
     * 清理对话资源
     */
    public void cleanupConversation(String conversationId) {
        conversationStates.remove(conversationId);
        AsrConnection asrConn = activeAsrConnections.remove(conversationId);
        if (asrConn != null) {
            asrConn.close();
        }
        log.info("清理对话资源: {}", conversationId);
    }

    /**
     * 清理过期对话
     */
    private void cleanupExpiredConversation(String conversationId) {
        ConversationState state = conversationStates.get(conversationId);
        if (state != null && !state.isActive()) {
            cleanupConversation(conversationId);
            log.info("清理过期对话: {}", conversationId);
        }
    }

    // ============ ASR相关辅助方法 ============

    private void sendAsrConfig(WebSocket webSocket) throws Exception {
        JSONObject config = new JSONObject();

        JSONObject common = new JSONObject();
        common.put("app_id", xfyunAiConfig.getAppId());
        config.put("common", common);

        JSONObject business = new JSONObject();
        business.put("language", "zh_cn");
        business.put("domain", "iat");
        business.put("accent", "mandarin");
        business.put("vinfo", 1);
        business.put("vad_eos", 5000);
        business.put("dwa", "wpgs");
        business.put("ptt", 0);
        business.put("rlang", "zh-cn");
        business.put("nunum", 0);
        config.put("business", business);

        JSONObject data = new JSONObject();
        data.put("status", 0);
        data.put("format", "audio/L16;rate=16000");
        data.put("encoding", "raw");
        data.put("audio", "");
        config.put("data", data);

        webSocket.send(config.toString());
        log.debug("发送ASR配置");
    }

    private void sendAudioFrame(WebSocket webSocket, byte[] audioData) {
        try {
            JSONObject audioFrame = new JSONObject();
            JSONObject data = new JSONObject();
            data.put("status", 1);
            data.put("format", "audio/L16;rate=16000");
            data.put("encoding", "raw");
            data.put("audio", Base64Utils.encodeToString(audioData));

            audioFrame.put("data", data);
            webSocket.send(audioFrame.toString());

        } catch (Exception e) {
            log.error("发送音频帧失败", e);
        }
    }

    private void sendEndFrame(WebSocket webSocket) {
        try {
            JSONObject endFrame = new JSONObject();
            JSONObject endData = new JSONObject();
            endData.put("status", 2);
            endData.put("format", "audio/L16;rate=16000");
            endData.put("encoding", "raw");
            endData.put("audio", "");
            endFrame.put("data", endData);

            webSocket.send(endFrame.toString());
            log.info("发送ASR结束帧");

        } catch (Exception e) {
            log.error("发送结束帧失败", e);
        }
    }

    private String parseAsrResult(String result) {
        try {
            JSONObject resultObj = JSONUtil.parseObj(result);
            StringBuilder text = new StringBuilder();

            if (resultObj.containsKey("ws")) {
                JSONArray wsArray = resultObj.getJSONArray("ws");
                for (int i = 0; i < wsArray.size(); i++) {
                    JSONObject wsItem = wsArray.getJSONObject(i);
                    if (wsItem.containsKey("cw")) {
                        JSONArray cwArray = wsItem.getJSONArray("cw");
                        for (int j = 0; j < cwArray.size(); j++) {
                            JSONObject cwItem = cwArray.getJSONObject(j);
                            String w = cwItem.getStr("w");
                            if (w != null && !w.trim().isEmpty()) {
                                text.append(w);
                            }
                        }
                    }
                }
            }

            return text.toString().trim();
        } catch (Exception e) {
            log.warn("解析ASR结果失败", e);
            return "";
        }
    }

    private String buildAsrAuthUrl() throws Exception {
        SimpleDateFormat sdf = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss z", Locale.US);
        sdf.setTimeZone(TimeZone.getTimeZone("GMT"));
        String date = sdf.format(new Date());

        String host = "iat-api.xfyun.cn";
        String requestLine = "GET /v2/iat HTTP/1.1";
        String signatureOrigin = "host: " + host + "\n" + "date: " + date + "\n" + requestLine;

        Mac mac = Mac.getInstance("HmacSHA256");
        SecretKeySpec spec = new SecretKeySpec(xfyunAiConfig.getApiSecret().getBytes(StandardCharsets.UTF_8), "HmacSHA256");
        mac.init(spec);
        byte[] signed = mac.doFinal(signatureOrigin.getBytes(StandardCharsets.UTF_8));
        String signature = Base64Utils.encodeToString(signed);

        String authorizationOrigin = String.format(
                "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"",
                xfyunAiConfig.getApiKey(), "hmac-sha256", "host date request-line", signature
        );

        String authorization = Base64Utils.encodeToString(authorizationOrigin.getBytes(StandardCharsets.UTF_8));

        return String.format("%s?authorization=%s&date=%s&host=%s",
                xfyunAiConfig.getAsrWsUrl(),
                URLEncoder.encode(authorization, StandardCharsets.UTF_8.name()),
                URLEncoder.encode(date, StandardCharsets.UTF_8.name()),
                URLEncoder.encode(host, StandardCharsets.UTF_8.name())
        );
    }

    // ============ 音频格式转换方法 ============

    /**
     * 将PCM音频数据转换为WAV格式
     */
    private byte[] pcmToWav(byte[] pcmData, int sampleRate, int channels, int bitsPerSample) {
        try {
            ByteArrayOutputStream wavStream = new ByteArrayOutputStream();

            int dataLength = pcmData.length;
            int fileLength = dataLength + 36;

            // RIFF chunk
            wavStream.write("RIFF".getBytes());
            wavStream.write(intToByteArray(fileLength), 0, 4);
            wavStream.write("WAVE".getBytes());

            // fmt chunk
            wavStream.write("fmt ".getBytes());
            wavStream.write(intToByteArray(16), 0, 4);
            wavStream.write(shortToByteArray((short) 1), 0, 2);
            wavStream.write(shortToByteArray((short) channels), 0, 2);
            wavStream.write(intToByteArray(sampleRate), 0, 4);
            wavStream.write(intToByteArray(sampleRate * channels * bitsPerSample / 8), 0, 4);
            wavStream.write(shortToByteArray((short) (channels * bitsPerSample / 8)), 0, 2);
            wavStream.write(shortToByteArray((short) bitsPerSample), 0, 2);

            // data chunk
            wavStream.write("data".getBytes());
            wavStream.write(intToByteArray(dataLength), 0, 4);
            wavStream.write(pcmData);

            return wavStream.toByteArray();
        } catch (Exception e) {
            log.error("PCM转WAV失败", e);
            return pcmData;
        }
    }

    /**
     * 整数转字节数组（小端）
     */
    private byte[] intToByteArray(int value) {
        return new byte[] {
                (byte) (value & 0xff),
                (byte) ((value >> 8) & 0xff),
                (byte) ((value >> 16) & 0xff),
                (byte) ((value >> 24) & 0xff)
        };
    }

    /**
     * 短整数转字节数组（小端）
     */
    private byte[] shortToByteArray(short value) {
        return new byte[] {
                (byte) (value & 0xff),
                (byte) ((value >> 8) & 0xff)
        };
    }

    // ============ 客户端通信方法（添加同步控制） ============

    private void sendAsrResultToClient(WebSocketSession session, String conversationId,
                                       String text, boolean isFinal) {
        synchronized (getSessionLock(session)) {
            try {
                if (!session.isOpen()) {
                    log.warn("WebSocket会话已关闭，跳过发送ASR结果");
                    return;
                }

                JSONObject message = new JSONObject();
                message.put("type", "asr_result");
                message.put("conversationId", conversationId);
                message.put("text", text);
                message.put("isFinal", isFinal);
                message.put("confidence", 1.0);
                message.put("timestamp", System.currentTimeMillis());

                session.sendMessage(new TextMessage(message.toString()));
                log.debug("发送ASR结果: '{}', isFinal: {}", text, isFinal);

            } catch (Exception e) {
                log.error("发送ASR结果失败", e);
            }
        }
    }

    private void sendUserInputCompleteToClient(WebSocketSession session, String conversationId, String text) {
        synchronized (getSessionLock(session)) {
            try {
                if (!session.isOpen()) {
                    return;
                }

                JSONObject message = new JSONObject();
                message.put("type", "user_input_complete");
                message.put("conversationId", conversationId);
                message.put("text", text);
                message.put("timestamp", System.currentTimeMillis());

                session.sendMessage(new TextMessage(message.toString()));
            } catch (Exception e) {
                log.error("发送用户输入完成信号失败", e);
            }
        }
    }

    private void sendTextChunkToClient(WebSocketSession session, String conversationId,
                                       String messageId, String chunk) {
        synchronized (getSessionLock(session)) {
            try {
                if (!session.isOpen()) {
                    return;
                }

                JSONObject message = new JSONObject();
                message.put("type", "ai_text_chunk");
                message.put("conversationId", conversationId);
                message.put("messageId", messageId);
                message.put("chunk", chunk);
                message.put("timestamp", System.currentTimeMillis());

                session.sendMessage(new TextMessage(message.toString()));
            } catch (Exception e) {
                log.error("发送AI文本chunk失败", e);
            }
        }
    }

    /**
     * 发送音频chunk（同步控制，转换为WAV格式）
     */
    private void sendAudioChunkToClient(WebSocketSession session, String conversationId,
                                        String messageId, byte[] audioData) {
        synchronized (getSessionLock(session)) {
            try {
                if (audioData == null || audioData.length == 0) {
                    return;
                }

                if (!session.isOpen()) {
                    log.warn("WebSocket会话已关闭，跳过发送音频chunk");
                    return;
                }

                // 关键：将PCM转换为WAV格式
                byte[] wavData = pcmToWav(audioData, 16000, 1, 16);

                JSONObject message = new JSONObject();
                message.put("type", "ai_audio_chunk");
                message.put("conversationId", conversationId);
                message.put("messageId", messageId);
                message.put("audioData", Base64.getEncoder().encodeToString(wavData));
                message.put("audioSize", wavData.length);
                message.put("audioFormat", "wav");
                message.put("timestamp", System.currentTimeMillis());

                session.sendMessage(new TextMessage(message.toString()));
                log.debug("发送音频chunk: {} bytes (WAV格式)", wavData.length);

            } catch (Exception e) {
                log.error("发送音频chunk失败", e);
            }
        }
    }

    private void sendAiResponseCompleteToClient(WebSocketSession session, String conversationId, String messageId) {
        synchronized (getSessionLock(session)) {
            try {
                if (!session.isOpen()) {
                    return;
                }

                JSONObject message = new JSONObject();
                message.put("type", "ai_response_complete");
                message.put("conversationId", conversationId);
                message.put("messageId", messageId);
                message.put("timestamp", System.currentTimeMillis());

                session.sendMessage(new TextMessage(message.toString()));
            } catch (Exception e) {
                log.error("发送AI回复完成信号失败", e);
            }
        }
    }

    private void sendErrorToClient(WebSocketSession session, String error) {
        synchronized (getSessionLock(session)) {
            try {
                if (!session.isOpen()) {
                    return;
                }

                JSONObject message = new JSONObject();
                message.put("type", "error");
                message.put("error", error);
                message.put("timestamp", System.currentTimeMillis());

                session.sendMessage(new TextMessage(message.toString()));
            } catch (Exception e) {
                log.error("发送错误信息失败", e);
            }
        }
    }

    // ============ 内部类 ============

    private static class VoiceStreamBuffer {
        private final WebSocketSession session;
        private final String conversationId;
        private final ByteArrayOutputStream buffer = new ByteArrayOutputStream();
        private long lastActivityTime = System.currentTimeMillis();

        private static final int BUFFER_SIZE_THRESHOLD = 16384;
        private static final long SILENCE_TIMEOUT_MS = 1000;

        public VoiceStreamBuffer(WebSocketSession session, String conversationId) {
            this.session = session;
            this.conversationId = conversationId;
        }

        public synchronized void addAudioData(byte[] data) {
            buffer.write(data, 0, data.length);
            lastActivityTime = System.currentTimeMillis();
        }

        public synchronized byte[] getAndClearBuffer() {
            byte[] data = buffer.toByteArray();
            buffer.reset();
            return data;
        }

        public boolean shouldProcess() {
            long currentTime = System.currentTimeMillis();
            return buffer.size() >= BUFFER_SIZE_THRESHOLD ||
                    (buffer.size() > 2048 && (currentTime - lastActivityTime) > SILENCE_TIMEOUT_MS);
        }

        public WebSocketSession getSession() {
            return session;
        }

        public String getConversationId() {
            return conversationId;
        }

        public void cleanup() {
            try {
                buffer.close();
            } catch (Exception e) {
                // ignore
            }
        }
    }

    private static class ConversationState {
        private String conversationId;
        private String roleId;
        private String userId;
        private boolean active;
        private boolean recording = false;
        private StringBuilder asrTextBuilder = new StringBuilder();

        public String getConversationId() {
            return conversationId;
        }

        public void setConversationId(String conversationId) {
            this.conversationId = conversationId;
        }

        public String getRoleId() {
            return roleId;
        }

        public void setRoleId(String roleId) {
            this.roleId = roleId;
        }

        public String getUserId() {
            return userId;
        }

        public void setUserId(String userId) {
            this.userId = userId;
        }

        public boolean isActive() {
            return active;
        }

        public void setActive(boolean active) {
            this.active = active;
        }

        public boolean isRecording() {
            return recording;
        }

        public void setRecording(boolean recording) {
            this.recording = recording;
        }

        public String getAccumulatedText() {
            return asrTextBuilder.toString();
        }

        public void appendAsrText(String text) {
            if (text != null && !text.trim().isEmpty()) {
                asrTextBuilder.append(text);
            }
        }

        public void clearAsrText() {
            asrTextBuilder.setLength(0);
        }
    }

    private static class AsrConnection {
        private WebSocket webSocket;
        private StringBuilder textBuilder = new StringBuilder();
        private boolean isActive = true;

        public WebSocket getWebSocket() {
            return webSocket;
        }

        public void setWebSocket(WebSocket ws) {
            this.webSocket = ws;
        }

        // 原有的appendText方法保持不变（用于其他场景）
        public void appendText(String text) {
            if (text != null && !text.trim().isEmpty()) {
                textBuilder.append(text);
            }
        }

        // 新增：直接设置文本，不累积（用于处理讯飞ASR的累积结果）
        public void setText(String text) {
            if (text != null) {
                textBuilder.setLength(0); // 清空
                textBuilder.append(text.trim());
            }
        }

        public String getFinalText() {
            return textBuilder.toString().trim();
        }

        public void clearText() {
            textBuilder.setLength(0);
        }

        public boolean isActive() {
            return isActive;
        }

        public void setActive(boolean active) {
            this.isActive = active;
        }

        public void close() {
            isActive = false;
            if (webSocket != null) {
                webSocket.close(1000, "Normal closure");
            }
        }
    }

    public interface AsrStreamCallback {
        void onResult(String text, boolean isFinal);

        void onError(String error);
    }

    public interface TtsStreamCallback {
        void onAudioChunk(byte[] audioData);

        void onComplete();

        void onError(String error);
    }
}